package com.shujia.flink.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo2SourceFunction {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //使用自定义的source
    val myDS: DataStream[Int] = env.addSource(new MySourceFunction)


    myDS.print()

    env.execute()

  }

}


class MySourceFunction extends SourceFunction[Int] {
  /**
    *
    * run 方法是在任务启动的时候执行一次
    *
    * 生成数据的方法，将拿到的数据发送到下游
    *
    * @param ctx ： 用于将数据发送到下游
    */
  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {


    var i = 0

    while (true) {

      //将数据发送到下游
      ctx.collect(i)

      Thread.sleep(100)

      i += 1
    }


  }

  //用于回收资源的
  //在取消任务的时候执行一次
  override def cancel(): Unit = {

  }
}